use std::cell::RefCell;
use std::collections::HashMap;
use std::io::{self, Read, Write};
use std::net::SocketAddr;
use std::rc::Rc;
// use std::cell::RefMut;
use std::time::{Duration};

// use std::future::AsyncWrite;
use mio::event::Evented;
use mio::{self, Ready, Registration, PollOpt, Token, SetReadiness};

use std::task::{Poll, Context};
use std::pin::Pin;
use std::future::Future;

use tokio::net::UdpSocket;
// use std::net::UdpSocket;
use tokio::io::AsyncWrite;
use tokio::io::AsyncWriteExt;
use tokio::io::AsyncRead;
use tokio::io::AsyncReadExt;
use tokio::io::PollEvented;
use tokio::time::timeout;
use tokio::time::timeout_at;
use tokio::time::Instant;
use tokio::time::Elapsed;
use tokio::time::Timeout;
use tokio::prelude::*;
// use futures::async_await::PollOnce;

use bytes::{Buf, BufMut, ByteOrder, LittleEndian};

use crate::kcp::kcb::Kcb;
use crate::kcp::ctime;

use crate::utils::Except;
use crate::utils::err_with;


macro_rules! ready_poll {
    ($future:expr, $cx:expr) => {
        {
            
            let mut f = Box::pin($future);
            let futur = Pin::new(&mut f);
            
            futur.poll($cx)
        }
    }
}


#[allow(unused)]
struct KcpPair<'a>{
    k:Rc<RefCell<Kcb<KcpOutput>>>,
    set_readiness: SetReadiness,
    token: Rc<RefCell<&'a mut Timeout<Nothing>>>,
}

#[allow(unused)]
pub struct KcpListener<'a> {
    udp: Rc<RefCell<UdpSocket>>,
    connections: HashMap<SocketAddr, KcpPair<'a>>
}

#[allow(unused)]
impl <'a>KcpListener<'a>{
    pub async fn bind(addr: &SocketAddr) -> io::Result<KcpListener<'a>>{
        match UdpSocket::bind(addr).await{
            Ok(udp) => {
                Ok(KcpListener{
                    udp:Rc::new(RefCell::new(udp)),
                    connections: HashMap::new(),
                })
            },
            Err(e) => Result::Err(e)
        }
        
    }

    pub async fn accept(&mut self) -> io::Result<(KcpStream, SocketAddr)> {
        let mut buf = [0;1024];
        loop{
            // tokio::spawn(async move{
            match self.udp.borrow_mut().recv_from(&mut buf).await{
                Err(e) => {
                    return Err(e);
                },
                Ok((n, addr)) => {
                    if self.connections.contains_key(&addr){
                        if let Some(mut kp) = self.connections.get_mut(&addr){
                            let mut kcb = kp.k.borrow_mut();
                            kcb.input(&buf[..n]);
                            kcb.update(clock());
                            let dur = kcb.check(clock());
                            let next = Instant::now() + Duration::from_millis(dur as u64);
                            let new_token:&'static mut Timeout<Nothing> = Box::leak(Box::new(timeout_at(next, Nothing)));
                            
                            // reset timeout
                            kp.token = Rc::new(RefCell::new(new_token));
                            
                            kp.set_readiness.set_readiness(mio::Ready::readable());

                            
                        }
                    }else{
                        // this is new accept udp socket.
                        let conv = LittleEndian::read_u32(&buf[..4]);
                        // create kcb protocoal struct . 
                        let mut kcb = Kcb::new(conv, KcpOutput{
                            udp: self.udp.clone(),
                            peer: addr.clone(),
                        });
                        
                        // init config for kcp 's windows size
                        kcb.wndsize(128, 128);

                        // init nodelay config 
                        kcb.nodelay(0, 10, 0, true);
                        let kcb = Rc::new(RefCell::new(kcb));
                        let (registration , set_readiness) = Registration::new2();
                        
                        // init token for timeout.
                        let now = Instant::now();
                        let next = now + Duration::from_millis(10);
                        let token_raw = timeout_at(next, Nothing);
                        let token = Rc::new(RefCell::new(token_raw));
                        // create core 
                        let core = KcpCore{
                            kcb: kcb.clone(),
                            registration: registration,
                            set_readiness: set_readiness.clone(),
                            token: token.clone(),
                        };
                        let shared_kcb = kcb.clone();
                        
                        let now = Instant::now();
                        let next = now + Duration::from_millis(10);
                        let token_static:&'static  mut  Timeout<_> = Box::leak(Box::new(timeout_at(next, Nothing)));
                        let interval = KcpInterval{
                            kcb: shared_kcb,
                            token: Rc::new(RefCell::new(token_static))
                        };

                        interval.await;   
                        
                        if let Ok(io) = PollEvented::new(core){
                            let stream = KcpStream{
                                io:io
                            };
                            stream.io.get_ref().kcb.borrow_mut().input(&buf[..n]);
                        }else{
                            panic!("create in poll evented from KcpCore Error!")
                        }
                        

                        

                        
                    }
                }
            }
            // })
        }
    }
}

#[allow(unused)]
pub struct KcpOutput {
    udp: Rc<RefCell<UdpSocket>>,
    peer: SocketAddr,
}


#[allow(unused)]
impl KcpOutput {
    pub async fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
        let mut u = self.udp.borrow_mut();
        u.send_to(buf, &self.peer).await
    }

    pub async fn flush(&mut self) -> io::Result<()> {
        Ok(())
    }


    pub async fn write_all(&mut self, buf: &[u8]){
        let mut now_write:usize = 0;
        let need_write = buf.len();
        loop{
            now_write +=  self.write(buf).await.unwrap();
            if now_write == need_write{break}
        }
    }

}

impl AsyncReadExt for KcpOutput {
    fn poll_read(self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
        let mut reader = self.udp.borrow_mut();
        ready_poll!(reader.recv(&mut buf),cx)
    }
}

impl AsyncWrite for KcpOutput {
    fn poll_write(self: Pin<&mut Self>,cx: &mut Context<'_>, buf: &[u8]) -> Poll<io::Result<usize>> {
        let mut write = self.udp.borrow_mut();
        ready_poll!(write.send_to(buf, &self.peer), cx)
    }

    #[inline]
    fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
        // tcp flush is a no-op
        Poll::Ready(Ok(()))
    }

    fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
        Poll::Ready(Ok(().into()))
    }

}

// impl Write for KcpOutput{
//     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
//         let udp = self.udp.borrow_mut();
//         let u = tokio::spawn(async {
//             udp.send_to(buf, &self.peer).await
//         });
//         // u.join()
        
//     }

//     fn flush(&mut self) -> io::Result<()> {
//         Ok(())
//     }
// }

struct KcpInterval<'a, T> 
where T:Future
{
    kcb: Rc<RefCell<Kcb<KcpOutput>>>,
    token: Rc<RefCell<&'a mut Timeout<T>>>,
}
impl <'a>KcpInterval<'a, Nothing>{
    fn get_pin_mut_token(&mut self, ctx:&mut Context)-> Poll<Result<(), Elapsed>>{
        let mut token_mut = self.token.borrow_mut();
        let t = Pin::new(&mut *token_mut);
        t.poll(ctx)
        // Pin::new(&mut *token_mut)
    }
    fn kcp_check(&mut self, ctx:&mut Context) -> Poll::<Instant>{
        let mut kcb = self.kcb.borrow_mut();
        // kcb.update(clock());
        // let dur = self.kcb.borrow().check(clock());
        // let next = Instant::now() + Duration::from_millis(dur as u64);
        // Poll::Ready(next)

        let mut pin_future = Box::pin(kcb.update(clock()));
        let f = Pin::new(&mut pin_future);
        match f.poll(ctx){
            Poll::Ready(_) => {
                let dur = self.kcb.borrow().check(clock());
                let next = Instant::now() + Duration::from_millis(dur as u64);
                Poll::Ready(next)
            },
            Poll::Pending => Poll::Pending
        }
        
        
    }

}
#[derive(Clone)]
struct Nothing;
impl Future for Nothing {
    type Output = ();
    fn poll(self: Pin<&mut Self>, _: &mut Context) -> Poll<Self::Output> {
        Poll::Ready(())
    }
}



impl <'a>Future for KcpInterval<'a, Nothing>{
    type Output = Result<Nothing, Elapsed>;

    fn poll(mut self: Pin<&mut Self>, ctx: &mut Context) -> Poll<Self::Output> {
        match self.as_mut().get_pin_mut_token(ctx){
            Poll::Ready(test_elased) => {
                match test_elased{
                    Ok(_) => {    
                        match self.as_mut().kcp_check(ctx){
                            Poll::Ready(next) => {
                                let new_token:&'static mut Timeout<Nothing> = Box::leak(Box::new(timeout_at(next, Nothing)));
                                self.as_mut().token = Rc::new(RefCell::new(new_token));
                                Poll::Ready(Ok(Nothing))
                            },
                            Poll::Pending => Poll::Pending
                        }
                        
                    },
                    Err(e) => Poll::Ready(Err(e))
                }
            },
            Poll::Pending => Poll::Pending,
        }
    }
}

struct KcpCore {
    kcb: Rc<RefCell<Kcb<KcpOutput>>>,
    registration: Registration,
    set_readiness: SetReadiness,
    // cx: Context<'a>,
    token: Rc<RefCell<Timeout<Nothing>>>,
}

impl Read for KcpCore {
    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
        let result = self.kcb.borrow_mut().recv(buf);
        match result {
            Err(_) => Err(io::Error::new(io::ErrorKind::WouldBlock, "would block")),
            Ok(n) => Ok(n),
        }
    }
}



// impl <'a>Write for KcpCore<'a> {
//     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
//         let mut kcb = self.kcb.borrow_mut();
//         // let token = self.token.borrow_mut();
//         let result = kcb.send(buf);
        
//         while let Poll::Ready(_) = ready_poll!(kcb.update(clock()),&mut self.cx){
//             // wait poll loop
//         }
//         let dur = self.kcb.borrow().check(clock());
//         let mut kcb = self.kcb.borrow_mut();
//         while let Poll::Ready(_) = ready_poll!(kcb.flush(), &mut self.cx){
//             // wait poll loop
//         };
//         self.token = Rc::new(RefCell::new(timeout_at(Instant::now() + Duration::from_millis(dur as u64), Nothing)));
//         result
//     }

//     fn flush(&mut self) -> io::Result<()> {
//         Ok(())
//     }
// }

// impl Write for KcpCore {
//     fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
//         let mut kcb = self.kcb.borrow_mut();
//         // let token = self.token.borrow_mut();
//         let result = kcb.send(buf);
        
//         kcb.update(clock());
//         let dur = self.kcb.borrow().check(clock());
//         let mut kcb = self.kcb.borrow_mut();
//         kcb.flush();
//             // wait poll loop
        
//         self.token = Rc::new(RefCell::new(timeout_at(Instant::now() + Duration::from_millis(dur as u64), Nothing)));
//         result
//     }

//     fn flush(&mut self) -> io::Result<()> {
//         Ok(())
//     }
// }

impl AsyncRead for KcpCore {
    unsafe fn prepare_uninitialized_buffer(&self, _: &mut [std::mem::MaybeUninit<u8>]) -> bool {
        false
    }
    fn poll_read(self: Pin<&mut Self>, cx: &mut Context, buf: &[u8]) -> Poll<io::Result<usize>>{
        let mut kcb = self.kcb.borrow_mut();
        // let mut read_future = self.io.read(buf);
        ready_poll!(kcb.read(&mut buf), cx)
    }
}
    
impl AsyncWrite for KcpCore{
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<io::Result<usize>> {

        let mut kcb = self.kcb.borrow_mut();
        let token = self.token.borrow_mut();
        let result = kcb.send(buf);
        
        match ready_poll!(kcb.update(clock()),cx){
            Poll::Ready(_) =>{
                let dur = self.kcb.borrow().check(clock());
                let mut kcb = self.kcb.borrow_mut();
                match ready_poll!(kcb.flush(),cx){
                    Poll::Ready(_) => {
                        self.token = Rc::new(RefCell::new(timeout_at(Instant::now() + Duration::from_millis(dur as u64), Nothing)));
                        Poll::Ready(result)
                    },
                    Poll::Pending => Poll::Pending
                }
            },
        }
    }

    #[inline]
    fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
        // tcp flush is a no-op
        Poll::Ready(Ok(()))
    }

    fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
        Poll::Ready(Ok(().into()))
    }
}
// impl <'a>KcpCore<'a> {
//     async fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
//         let mut kcb = self.kcb.borrow_mut();
//         // let token = self.token.borrow_mut();
//         let result = kcb.send(buf);
        
//         kcb.update(clock()).await;
//         let dur = self.kcb.borrow().check(clock());
//         let mut kcb = self.kcb.borrow_mut();
//         kcb.flush().await;
//         self.token = Rc::new(RefCell::new(timeout_at(Instant::now() + Duration::from_millis(dur as u64), Nothing)));
//         result
//     }

//     fn flush(&mut self) -> io::Result<()> {
//         Ok(())
//     }
// }

impl Evented for KcpCore {
    fn register(&self, poll: &mio::Poll, token: Token,
        interest: Ready, opts: PollOpt) -> io::Result<()> {
        
        // self.kcb.register(poll, token, interest, opts)
        self.registration.register(poll, token, interest, opts)
    }

    fn reregister(&self, poll: &mio::Poll, token: Token,
            interest: Ready, opts: PollOpt) -> io::Result<()> {
        
        self.registration.reregister(poll, token, interest, opts)
    }

    fn deregister(&self, poll: &mio::Poll) -> io::Result<()> {
        // self.registration.deregister(poll)
        poll.deregister(self)
    }
}


pub struct KcpStream {
    // io: PollEvented<KcpCore>,
    io:KcpCore
}



impl KcpStream{
    // pub(crate) fn poll_write_priv(
    //     &mut self,
    //     cx: &mut Context<'_>,
    //     buf: &[u8],
    // ) -> Poll<io::Result<usize>> {
    //     ready!(self.io.poll_write_ready(cx))?;

    //     let _ = self.io.get_ref().set_readiness.set_readiness(
    //         mio::Ready::writable(),
    //     );
        
    //     // self.io.write(buf).await;
    //     let mut write_future = Box::pin(self.io.write(buf));
    //     match Pin::new(&mut write_future).poll(cx){
    //         Poll::Ready(result) => {
    //             match result {
    //                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
    //                     self.io.clear_write_ready(cx)?;
    //                     Poll::Pending
    //                 },
    //                 x => Poll::Ready(x),
    //             }
    //         },
    //         Poll::Pending => Poll::Pending
    //     }
    // }

    // pub(crate) fn poll_read_priv(
    //     &mut self,
    //     cx: &mut Context<'_>,
    //     buf: &mut [u8],
    // ) -> Poll<io::Result<usize>> {
    //     ready!(self.io.poll_read_ready(cx, mio::Ready::readable()))?;
    //     let mut read_future = self.io.read(buf);
    //     match Pin::new(&mut read_future).poll(cx){
    //         Poll::Ready(result) => {
    //             match result {
    //                 Err(ref e) if e.kind() == io::ErrorKind::WouldBlock => {
    //                     self.io.clear_read_ready(cx, mio::Ready::readable())?;
    //                     Poll::Pending
    //                 },
    //                 x => Poll::Ready(x),
    //             }
    //         },
    //         Poll::Pending => Poll::Pending
    //     }
    // }
}

impl AsyncRead for KcpStream {
    unsafe fn prepare_uninitialized_buffer(&self, _: &mut [std::mem::MaybeUninit<u8>]) -> bool {
        false
    }

    fn poll_read(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &mut [u8],
    ) -> Poll<io::Result<usize>> {
        ready_poll!(self.io.read_all(&mut buf), cx)
    }
}

impl AsyncWrite for KcpStream {
    fn poll_write(
        self: Pin<&mut Self>,
        cx: &mut Context<'_>,
        buf: &[u8],
    ) -> Poll<io::Result<usize>> {
        // let write = self.get_mut();
        // let mut write_future = Box::pin(self.io.write(buf));
        ready_poll!(self.io.write_all(&mut buf), cx)
    }

    #[inline]
    fn poll_flush(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
        // tcp flush is a no-op
        Poll::Ready(Ok(()))
    }

    fn poll_shutdown(self: Pin<&mut Self>, _: &mut Context<'_>) -> Poll<io::Result<()>> {
        Poll::Ready(Ok(().into()))
    }
}


#[inline]
fn clock() -> u32 {
    let timespec = ctime::get_time();
    let mills = timespec.sec * 1000 + timespec.nsec as i64 / 1000 / 1000;
    mills as u32
}

